package com.one2manycall.service.websocket;


import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.google.gson.JsonObject;
import com.one2manycall.service.UserOnlineManageServiceI;
import com.one2manycall.vo.ChatDataVo;
import com.one2manycall.vo.User;
import org.kurento.client.*;
import org.kurento.jsonrpc.JsonUtils;
import com.one2manycall.vo.UserSession;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketSession;

import java.io.IOException;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

/**
 * 消息处理类，针对不同的消息进行处理
 */
@Component
public class MessageHandler {


    private static final String RECORDER_FILE_PATH = "file:///tmp/HelloWorldRecorded.webm";



    private static final Logger LOGGER = LoggerFactory.getLogger(MessageHandler.class);

    public final ConcurrentHashMap<String, CallMediaPipeline> pipelines = new ConcurrentHashMap<>();


    private final ConcurrentHashMap<String, MediaPipeline> pipelinesMedia = new ConcurrentHashMap<>();


    @Autowired
    private UserRegistry registry;


    @Autowired
    private KurentoClient kurento;


    public void onIceCandidateHander(JsonObject jsonMessage, UserSession user) {
        JsonObject candidate = jsonMessage.get("candidate").getAsJsonObject();
        if (user != null) {
            IceCandidate cand =
                    new IceCandidate(candidate.get("candidate").getAsString(), candidate.get("sdpMid")
                            .getAsString(), candidate.get("sdpMLineIndex").getAsInt());
            user.addCandidate(cand);
        }
    }

    public void callTextHander(WebSocketSession session, JsonObject jsonMessage, UserSession user) throws IOException {
        try {
            LOGGER.info("all");
            call(user, jsonMessage);
        } catch (Throwable t) {
            handleErrorResponse(t, session, "callResponse");
        }
    }

    /**
     * 注册消息处理
     *
     * @param session     session
     * @param jsonMessage jsonMessage
     * @throws IOException
     */
    public void registerTextHander(WebSocketSession session, JsonObject jsonMessage) throws IOException {
        try {
            LOGGER.info("register");
            register(session, jsonMessage);
        } catch (Throwable t) {
            handleErrorResponse(t, session, "registerResponse");
        }
    }


    private void register(WebSocketSession session, JsonObject jsonMessage) throws IOException {

        LOGGER.info("register", jsonMessage);

        String name = jsonMessage.getAsJsonPrimitive("name").getAsString();

        UserSession caller = new UserSession(session, name);
        String responseMsg = "accepted";
        if (name.isEmpty()) {
            responseMsg = "rejected: empty user name";
        } else if (registry.exists(name)) {

            responseMsg = "rejected: user '" + name + "' already registered";
            LOGGER.info(responseMsg);
        } else {
            registry.register(caller);
        }

        JsonObject response = new JsonObject();
        response.addProperty("id", "registerResponse");
        response.addProperty("response", responseMsg);
        caller.sendMessage(response);
    }


    private void call(UserSession caller, JsonObject jsonMessage) throws IOException {
        String to = jsonMessage.get("to").getAsString();
        String from = jsonMessage.get("from").getAsString();
        JsonObject response = new JsonObject();

        if (registry.exists(to)) {
            caller.setSdpOffer(jsonMessage.getAsJsonPrimitive("sdpOffer").getAsString());
            caller.setCallingTo(to);

            response.addProperty("id", "incomingCall");
            response.addProperty("from", from);

            UserSession callee = registry.getByName(to);
            callee.sendMessage(response);
            callee.setCallingFrom(from);
        } else {
            response.addProperty("id", "callResponse");
            response.addProperty("response", "rejected: user '" + to + "' is not registered");

            caller.sendMessage(response);
        }
    }


    /**
     *  处理异常消息
     * @param throwable t
     * @param session s
     * @param responseId re
     * @throws IOException
     */
    public void handleErrorResponse(Throwable throwable, WebSocketSession session, String responseId)
            throws IOException {
        stop(session);
        LOGGER.error(throwable.getMessage(), throwable);
        JsonObject response = new JsonObject();
        response.addProperty("id", responseId);
        response.addProperty("response", "rejected");
        response.addProperty("message", throwable.getMessage());
        session.sendMessage(new TextMessage(response.toString()));
    }


    public void incomingCallResponse(final UserSession callee, JsonObject jsonMessage)
            throws IOException {
        String callResponse = jsonMessage.get("callResponse").getAsString();
        String from = jsonMessage.get("from").getAsString();
        final UserSession calleer = registry.getByName(from);
        String to = calleer.getCallingTo();

        if ("accept".equals(callResponse)) {
            LOGGER.debug("Accepted call from '{}' to '{}'", from, to);

            CallMediaPipeline pipeline = null;
            try {
                pipeline = new CallMediaPipeline(kurento);
                pipelines.put(calleer.getSessionId(), pipeline);
                pipelines.put(callee.getSessionId(), pipeline);

                callee.setWebRtcEndpoint(pipeline.getCalleeWebRtcEp());
                pipeline.getCalleeWebRtcEp().addIceCandidateFoundListener(
                        new EventListener<IceCandidateFoundEvent>() {

                            @Override
                            public void onEvent(IceCandidateFoundEvent event) {
                                JsonObject response = new JsonObject();
                                response.addProperty("id", "iceCandidate");
                                response.add("candidate", JsonUtils.toJsonObject(event.getCandidate()));
                                try {
                                    synchronized (callee.getSession()) {
                                        callee.getSession().sendMessage(new TextMessage(response.toString()));
                                    }
                                } catch (IOException e) {
                                    LOGGER.debug(e.getMessage());
                                }
                            }
                        });

                calleer.setWebRtcEndpoint(pipeline.getCallerWebRtcEp());
                pipeline.getCallerWebRtcEp().addIceCandidateFoundListener(
                        new EventListener<IceCandidateFoundEvent>() {

                            @Override
                            public void onEvent(IceCandidateFoundEvent event) {
                                JsonObject response = new JsonObject();
                                response.addProperty("id", "iceCandidate");
                                response.add("candidate", JsonUtils.toJsonObject(event.getCandidate()));
                                try {
                                    synchronized (calleer.getSession()) {
                                        calleer.getSession().sendMessage(new TextMessage(response.toString()));
                                    }
                                } catch (IOException e) {
                                    LOGGER.debug(e.getMessage());
                                }
                            }
                        });

                String calleeSdpOffer = jsonMessage.get("sdpOffer").getAsString();
                String calleeSdpAnswer = pipeline.generateSdpAnswerForCallee(calleeSdpOffer);
                JsonObject startCommunication = new JsonObject();
                startCommunication.addProperty("id", "startCommunication");
                startCommunication.addProperty("sdpAnswer", calleeSdpAnswer);

                synchronized (callee) {
                    callee.sendMessage(startCommunication);
                }

                pipeline.getCalleeWebRtcEp().gatherCandidates();

                String callerSdpOffer = registry.getByName(from).getSdpOffer();
                String callerSdpAnswer = pipeline.generateSdpAnswerForCaller(callerSdpOffer);
                JsonObject response = new JsonObject();
                response.addProperty("id", "callResponse");
                response.addProperty("response", "accepted");
                response.addProperty("sdpAnswer", callerSdpAnswer);

                synchronized (calleer) {
                    calleer.sendMessage(response);
                }

                pipeline.getCallerWebRtcEp().gatherCandidates();

            } catch (Throwable t) {
                LOGGER.error(t.getMessage(), t);

                if (pipeline != null) {
                    pipeline.release();
                }

                pipelines.remove(calleer.getSessionId());
                pipelines.remove(callee.getSessionId());

                JsonObject response = new JsonObject();
                response.addProperty("id", "callResponse");
                response.addProperty("response", "rejected");
                calleer.sendMessage(response);

                response = new JsonObject();
                response.addProperty("id", "stopCommunication");
                callee.sendMessage(response);
            }

        } else {
            JsonObject response = new JsonObject();
            response.addProperty("id", "callResponse");
            response.addProperty("response", "rejected");
            calleer.sendMessage(response);
        }
    }


    /**
     * 结束一个会话
     *
     * @param session session
     * @throws IOException
     */
    public void stop(WebSocketSession session) throws IOException {
        String sessionId = session.getId();
        if (pipelines.containsKey(sessionId)) {
            pipelines.get(sessionId).release();
            CallMediaPipeline pipeline = pipelines.remove(sessionId);
            pipeline.release();

            // Both users can stop the communication. A 'stopCommunication'
            // message will be sent to the other peer.
            UserSession stopperUser = registry.getBySession(session);
            if (stopperUser != null) {
                UserSession stoppedUser =
                        (stopperUser.getCallingFrom() != null) ? registry.getByName(stopperUser
                                .getCallingFrom()) : stopperUser.getCallingTo() != null ? registry
                                .getByName(stopperUser.getCallingTo()) : null;

                if (stoppedUser != null) {
                    JsonObject message = new JsonObject();
                    message.addProperty("id", "stopCommunication");
                    stoppedUser.sendMessage(message);
                    stoppedUser.clear();
                }
                stopperUser.clear();
            }

        }
    }


    public void play(final UserSession session, JsonObject jsonMessage) throws IOException {
        String user = jsonMessage.get("user").getAsString();
        LOGGER.debug("Playing recorded call of user '{}'", user);

        JsonObject response = new JsonObject();
        response.addProperty("id", "playResponse");

        if (registry.getByName(user) != null && registry.getBySession(session.getSession()) != null) {
            final PlayMediaPipeline playMediaPipeline =
                    new PlayMediaPipeline(kurento, user, session.getSession());

            session.setPlayingWebRtcEndpoint(playMediaPipeline.getWebRtc());

            playMediaPipeline.getPlayer().addEndOfStreamListener(new EventListener<EndOfStreamEvent>() {
                @Override
                public void onEvent(EndOfStreamEvent event) {
                    UserSession user = registry.getBySession(session.getSession());
                    releasePipeline(user);
                    playMediaPipeline.sendPlayEnd(session.getSession());
                }
            });

            playMediaPipeline.getWebRtc().addIceCandidateFoundListener(
                    new EventListener<IceCandidateFoundEvent>() {

                        @Override
                        public void onEvent(IceCandidateFoundEvent event) {
                            JsonObject response = new JsonObject();
                            response.addProperty("id", "iceCandidate");
                            response.add("candidate", JsonUtils.toJsonObject(event.getCandidate()));
                            try {
                                synchronized (session) {
                                    session.getSession().sendMessage(new TextMessage(response.toString()));
                                }
                            } catch (IOException e) {
                                LOGGER.debug(e.getMessage());
                            }
                        }
                    });

            String sdpOffer = jsonMessage.get("sdpOffer").getAsString();
            String sdpAnswer = playMediaPipeline.generateSdpAnswer(sdpOffer);

            response.addProperty("response", "accepted");

            response.addProperty("sdpAnswer", sdpAnswer);

            playMediaPipeline.play();
            pipelinesMedia.put(session.getSessionId(), playMediaPipeline.getPipeline());
            synchronized (session.getSession()) {
                session.sendMessage(response);
            }

            playMediaPipeline.getWebRtc().gatherCandidates();

        } else {
            response.addProperty("response", "rejected");
            response.addProperty("error", "No recording for user '" + user
                    + "'. Please type a correct user in the 'Peer' field.");
            session.getSession().sendMessage(new TextMessage(response.toString()));
        }
    }


    public void releasePipeline(UserSession session) {
        String sessionId = session.getSessionId();

        if (pipelines.containsKey(sessionId)) {
            pipelines.get(sessionId).release();
            pipelines.remove(sessionId);
        }
        session.setWebRtcEndpoint(null);
        session.setPlayingWebRtcEndpoint(null);

        // set to null the endpoint of the other user
        UserSession stoppedUser =
                (session.getCallingFrom() != null) ? registry.getByName(session.getCallingFrom())
                        : registry.getByName(session.getCallingTo());
        stoppedUser.setWebRtcEndpoint(null);
        stoppedUser.setPlayingWebRtcEndpoint(null);
    }


    @Autowired
    private UserOnlineManageServiceI userOnlineManageServiceI;



    /**
     *  把消息发送到列表里面
     * @param session sesion
     * @param jsonMessage json data
     */
    public void chat(final WebSocketSession session, JsonObject jsonMessage) {


        String token = jsonMessage.get("token").getAsString();

        String data = jsonMessage.get("data").getAsString();

        String id = session.getId();

        ChatDataVo chatDataVo = new ChatDataVo();

        chatDataVo.setMsg(data);

        chatDataVo.setSessionId(id);
        chatDataVo.setId("chatResponse");

        chatDataVo.setCreateTime(System.currentTimeMillis()+"");

        // TODO: 2024.12.29  补充用户名

        User byUserToken = userOnlineManageServiceI.findByUserToken(token);
        if (byUserToken != null){
            chatDataVo.setUserName(byUserToken.getUserName());
        }else {
            chatDataVo.setMsgType("notLogin");
            String s = JSON.toJSONString(chatDataVo);
            try {
                session.sendMessage(new TextMessage(s));
            } catch (IOException e) {
               LOGGER.error("error ",e);
            }

            return;

        }
        // 如果当前用户发送消息，但是还未注册过，在websocket session 里面注册

        UserSession bySession = registry.getBySession(session);
        if (bySession == null){
            UserSession userSession = new UserSession(session, chatDataVo.getUserName());
            registry.register(userSession);
        }

//        todo  数据入库
//        ChatDataHandler.list.add(chatDataVo);

        // 通知发送消息
        sendChatDataAll(chatDataVo);

    }


    /**
     *  把消息发送到每一个
     */
    public void sendChatDataAll(ChatDataVo chatDataVo) {

        ConcurrentMap<String, UserSession> all = registry.getAll();
        // 创建Gson对象
        for (UserSession userSession : all.values()) {
            try {

                WebSocketSession session = userSession.getSession();
                if (session == null) {
                    LOGGER.warn("this session is close");
                    continue;
                }
                userSession.sendChatDataVo(chatDataVo);

            } catch (IOException e) {
                LOGGER.error(e.getLocalizedMessage() + userSession.getSessionId());
            }
        }

    }



    public void start(final WebSocketSession session, JsonObject jsonMessage) {
        try {

            // 1. Media logic (webRtcEndpoint in loopback)
            MediaPipeline pipeline = kurento.createMediaPipeline();
            WebRtcEndpoint webRtcEndpoint = new WebRtcEndpoint.Builder(pipeline).build();
            webRtcEndpoint.connect(webRtcEndpoint);

            MediaProfileSpecType profile = getMediaProfileFromMessage(jsonMessage);

            RecorderEndpoint recorder = new RecorderEndpoint.Builder(pipeline, RECORDER_FILE_PATH)
                    .withMediaProfile(profile).build();

            // Error listeners.
            pipeline.addErrorListener(new EventListener<ErrorEvent>() {
                @Override
                public void onEvent(ErrorEvent ev) {
                    LOGGER.error("[MediaPipeline::ErrorEvent] Error code {}: '{}', source: {}, timestamp: {}, tags: {}, description: {}",
                            ev.getErrorCode(), ev.getType(), ev.getSource().getName(),
                            ev.getTimestampMillis(), ev.getTags(), ev.getDescription());
                    sendError(session, "[MediaPipeline] " + ev.getDescription());
                }
            });
            webRtcEndpoint.addErrorListener(new EventListener<ErrorEvent>() {
                @Override
                public void onEvent(ErrorEvent ev) {
                    LOGGER.error("[WebRtcEndpoint::ErrorEvent] Error code {}: '{}', source: {}, timestamp: {}, tags: {}, description: {}",
                            ev.getErrorCode(), ev.getType(), ev.getSource().getName(),
                            ev.getTimestampMillis(), ev.getTags(), ev.getDescription());
                    sendError(session, "[WebRtcEndpoint] " + ev.getDescription());
                }
            });
            recorder.addErrorListener(new EventListener<ErrorEvent>() {
                @Override
                public void onEvent(ErrorEvent ev) {
                    LOGGER.error("[RecorderEndpoint::ErrorEvent] Error code {}: '{}', source: {}, timestamp: {}, tags: {}, description: {}",
                            ev.getErrorCode(), ev.getType(), ev.getSource().getName(),
                            ev.getTimestampMillis(), ev.getTags(), ev.getDescription());
                    sendError(session, "[RecorderEndpoint] " + ev.getDescription());
                }
            });

            recorder.addRecordingListener(new EventListener<RecordingEvent>() {

                @Override
                public void onEvent(RecordingEvent event) {
                    JsonObject response = new JsonObject();
                    response.addProperty("id", "recording");
                    try {
                        synchronized (session) {
                            session.sendMessage(new TextMessage(response.toString()));
                        }
                    } catch (IOException e) {
                        LOGGER.error(e.getMessage());
                    }
                }

            });

            recorder.addStoppedListener(new EventListener<StoppedEvent>() {

                @Override
                public void onEvent(StoppedEvent event) {
                    JsonObject response = new JsonObject();
                    response.addProperty("id", "stopped");
                    try {
                        synchronized (session) {
                            session.sendMessage(new TextMessage(response.toString()));
                        }
                    } catch (IOException e) {
                        LOGGER.error(e.getMessage());
                    }
                }

            });

            recorder.addPausedListener(new EventListener<PausedEvent>() {

                @Override
                public void onEvent(PausedEvent event) {
                    JsonObject response = new JsonObject();
                    response.addProperty("id", "paused");
                    try {
                        synchronized (session) {
                            session.sendMessage(new TextMessage(response.toString()));
                        }
                    } catch (IOException e) {
                        LOGGER.error(e.getMessage());
                    }
                }

            });

            connectAccordingToProfile(webRtcEndpoint, recorder, profile);

            // 2. Store user session
            UserSession user = new UserSession(session);
            user.setMediaPipeline(pipeline);
            user.setWebRtcEndpoint(webRtcEndpoint);
            user.setRecorderEndpoint(recorder);
            registry.register(user);

            // 3. SDP negotiation
            String sdpOffer = jsonMessage.get("sdpOffer").getAsString();
            String sdpAnswer = webRtcEndpoint.processOffer(sdpOffer);

            // 4. Gather ICE candidates
            webRtcEndpoint.addIceCandidateFoundListener(new EventListener<IceCandidateFoundEvent>() {

                @Override
                public void onEvent(IceCandidateFoundEvent event) {
                    JsonObject response = new JsonObject();
                    response.addProperty("id", "iceCandidate");
                    response.add("candidate", JsonUtils.toJsonObject(event.getCandidate()));
                    try {
                        synchronized (session) {
                            session.sendMessage(new TextMessage(response.toString()));
                        }
                    } catch (IOException e) {
                        LOGGER.error(e.getMessage());
                    }
                }
            });

            JsonObject response = new JsonObject();
            response.addProperty("id", "startResponse");
            response.addProperty("sdpAnswer", sdpAnswer);

            synchronized (user) {
                session.sendMessage(new TextMessage(response.toString()));
            }

            webRtcEndpoint.gatherCandidates();

            recorder.record();
        } catch (Throwable t) {
            LOGGER.error("Start error", t);
            sendError(session, t.getMessage());
        }
    }

    private void sendError(WebSocketSession session, String message) {
        JsonObject response = new JsonObject();
        response.addProperty("id", "error");
        response.addProperty("message", message);

        try {
            synchronized (session) {
                session.sendMessage(new TextMessage(response.toString()));
            }
        } catch (IOException e) {
            LOGGER.error("Exception sending message", e);
        }
    }


    private MediaProfileSpecType getMediaProfileFromMessage(JsonObject jsonMessage) {

        MediaProfileSpecType profile;
        switch (jsonMessage.get("mode").getAsString()) {
            case "audio-only":
                profile = MediaProfileSpecType.WEBM_AUDIO_ONLY;
                break;
            case "video-only":
                profile = MediaProfileSpecType.WEBM_VIDEO_ONLY;
                break;
            default:
                profile = MediaProfileSpecType.WEBM;
        }

        return profile;
    }


    private void connectAccordingToProfile(WebRtcEndpoint webRtcEndpoint, RecorderEndpoint recorder,
                                           MediaProfileSpecType profile) {
        switch (profile) {
            case WEBM:
                webRtcEndpoint.connect(recorder, MediaType.AUDIO);
                webRtcEndpoint.connect(recorder, MediaType.VIDEO);
                break;
            case WEBM_AUDIO_ONLY:
                webRtcEndpoint.connect(recorder, MediaType.AUDIO);
                break;
            case WEBM_VIDEO_ONLY:
                webRtcEndpoint.connect(recorder, MediaType.VIDEO);
                break;
            default:
                throw new UnsupportedOperationException("Unsupported profile for this tutorial: " + profile);
        }
    }



}
